package com.socket.nio.muti.d1;

import com.socket.nio.util.ByteBufferPrinter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 多线程网络编程优化
 * <p>
 * 总结：利用队列解决两个线程之间传递的问题
 *
 * @author Hexudong
 * @date 2025/4/5
 */
@Slf4j
public class MutiThreadServer {

  public static void main(String[] args) throws IOException {
    Thread.currentThread().setName("boss");
    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.configureBlocking(false);
    ssc.bind(new InetSocketAddress(8082));

    Selector boss = Selector.open();
    ssc.register(boss, SelectionKey.OP_ACCEPT);

    Worker worker = new Worker("worker-0");
    //worker.register();
    while (true) {
      boss.select();
      Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
      while (iter.hasNext()) {
        SelectionKey key = iter.next();
        iter.remove();
        if (key.isAcceptable()) {
          SocketChannel sc = ssc.accept();
          sc.configureBlocking(false);
          log.debug("connected...{}", sc.getRemoteAddress());
          // 关联 worker 的 selector
          log.debug("before register...{}", sc.getRemoteAddress());
          worker.register(sc);
          log.debug("after register...{}", sc.getRemoteAddress());

        }
      }
    }
  }

  static class Worker implements Runnable {
    private Thread thread;
    private Selector selector;
    private String name;
    private volatile boolean start = false; // 是否初始化
    private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue();

    public Worker(String name) {
      this.name = name;
    }

    public void register(SocketChannel sc) throws IOException {
      if (!start) {
        selector = Selector.open();
        thread = new Thread(this, name);
        thread.start();
        start = true;
      }

      // 向队列中添加任务，并没有执行
      queue.add(() -> {
        try {
          sc.register(selector, SelectionKey.OP_READ, null); // boss
        } catch (ClosedChannelException e) {
          e.printStackTrace();
        }
      });

      // 唤醒下面的 select 方法
      selector.wakeup();
    }

    @Override
    public void run() {
      while (true) {
        try {
          selector.select(); // worker-0，该方法是一个阻塞方法

          Runnable task = queue.poll();
          if (task != null) {
            task.run();
          }

          Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
          while (iter.hasNext()) {
            SelectionKey key = iter.next();
            iter.remove();

            if (key.isReadable()) {
              ByteBuffer buffer = ByteBuffer.allocate(16);
              SocketChannel channel = (SocketChannel) key.channel();
              log.debug("read...{}", channel.getRemoteAddress());
              channel.read(buffer);
              buffer.flip();
              ByteBufferPrinter.prettyPrint(buffer);
            }
          }
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
    }
  }
}
